package bsf.baseservice.businessmq.redis;


import bsf.base.CallBack;
import bsf.baseservice.businessmq.systemruntime.SystemParamConfig;
import bsf.baseservice.businessmq.systemruntime.log.ErrorLogHelper;
import bsf.redis.RedisDb;
import bsf.redis.RedisManager;
import bsf.system.thread.ThreadPool;
import bsf.util.ConvertUtil;
import redis.clients.jedis.JedisPubSub;

import java.util.concurrent.ExecutorService;

/**
 Redis 网络命令监听器
 
*/
public class RedisNetCommandListener implements AutoCloseable
{
	public String RedisServerIp;
	private RedisDb redisDb; //当前监听连接
	private String mqPath;
	private String channelName;
	private boolean isdisposeing = false; //监听释放标记
	public String Name;

	public RedisNetCommandListener(String redisserverip)
	{
		RedisServerIp = redisserverip;
	}

	public final void register(final CallBack.Func2<Object,String,String> action, final String mqpath, final String channel)
	{
		mqPath = mqpath;
		channelName = channel;
		ThreadPool.SystemThreadPool.submit(new Runnable() { //开启及时网络订阅
			@Override
			public void run() {
				netSubscribe(action, mqpath, channel);
			}
		});
	}

	private void netSubscribe(CallBack.Func2<Object,String,String> action, String mqpath, String channel)
	{
		while (isdisposeing==false)
		{
			try
			{
				try
				{
					closeRedisClient();
					redisSubscribe(action, mqpath, channel);
				}
				catch (Exception exp)
				{
					if (isdisposeing == false)
					{
						ErrorLogHelper.writeLine(-1, mqpath, "NetSubscribe", String.format("MQ心跳redis订阅通信消息出错,请检查redis服务器,订阅名:%1$s", Name), exp);
					}
				}
				Thread.sleep(SystemParamConfig.Redis_Subscribe_FailConnect_ReConnect_Every_Time * 1000);
			}
			catch (Exception exp)
			{
				// when thread is sleeping,but we cancel this thread,may throw thread abort error 
			}
		}
	}

	private void redisSubscribe(CallBack.Func2<Object,String,String> action, String mqpath, String channelname)
	{
		RedisManager manager = new RedisManager();
		redisDb = manager.createClient(RedisServerIp.split("[:]", -1)[0], Integer.parseInt(RedisServerIp.split("[:]", -1)[1]), "");
		MyPubSub sub = new MyPubSub();sub.messagecallback = action;sub.redisDb = redisDb;sub.MqPath=mqpath;
		redisDb.getClient().subscribe(sub,channelname);
	}

private class MyPubSub extends JedisPubSub {
	public CallBack.Func2<Object,String,String> messagecallback;
	public RedisDb redisDb;
	public String MqPath;

	// 取得订阅的消息后的处理
	public void onMessage(String channel, String message) {
		try
		{
			if (message.equals("RedisNetCommandListener-Close") || isdisposeing==true) //关闭通道
			{
				try
				{
					/*subscription.UnSubscribeFromChannels(channel);*/
					this.unsubscribe();
				}
				catch (Exception e)
				{
				}
			}
			else
			{
				if (messagecallback != null)
				{
					messagecallback.invoke(channel, message);
				}
			}
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(-1, MqPath, "RedisSubscribe",
					String.format("MQredis订阅通信消息,通道:%2$s,处理消息%1$s,订阅名:%3$s出错", ConvertUtil.nullToStr(message), channel, Name), exp);
		}
	}

	// 初始化订阅时候的处理
	public void onSubscribe(String channel, int subscribedChannels) {
		//System.out.println(channel + "=" + subscribedChannels+"&&&&&&&");
	}

	// 取消订阅时候的处理
	public void onUnsubscribe(String channel, int subscribedChannels) {
		//System.out.println(channel + "=" + subscribedChannels+"#########");
	}

	// 初始化按表达式的方式订阅时候的处理
	public void onPSubscribe(String pattern, int subscribedChannels) {
		//System.out.println(pattern + "=" + subscribedChannels+"!!!!!!!");
	}

	// 取消按表达式的方式订阅时候的处理
	public void onPUnsubscribe(String pattern, int subscribedChannels) {
		//System.out.println(pattern + "=" + subscribedChannels+"@@@@@@@@@");
	}

	// 取得按表达式的方式订阅的消息后的处理
	public void onPMessage(String pattern, String channel, String message) {
		//System.out.println(pattern + "=" + channel + "=" + message);
	}
}
	private void closeRedisClient()
	{
		try
		{
			if (redisDb != null)
			{

				if (redisDb.getClient() != null)
				{
					redisDb.getClient().close();
				}
				//redisDb.Dispose();
				redisDb = null;
			}
		}
		catch (Exception exp)
		{
		}
	}

	public final void close()
	{
		isdisposeing = true;
		try
		{
			try
			{
				//通知关闭监听通道
				if (redisDb != null)
				{
					RedisManager manager = new RedisManager();
				   try(RedisDb db = manager.createClient(RedisServerIp.split("[:]", -1)[0], Integer.parseInt(RedisServerIp.split("[:]", -1)[1]), "")) {
						   db.getClient().publish(channelName, "RedisNetCommandListener-Close");
						   db.getClient().close();
				   }
				}
			}
			catch (Exception e)
			{
			}
			//关闭当前连接
			closeRedisClient();
		}
		catch (Exception exp)
		{
		}
	}
}
